#ifndef __M_BROKER_H__
#define __M_BROKER_H__

#include"mq_host.hpp"
#include"mq_connection.hpp"
#include"mq_consumer.hpp"
#include"../mqcommon/mq_threadPool.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_logger.hpp"

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"


namespace xiang11mq
{

class Broker
{
    #define DBFILE "/meta.db"
    #define HOSTNAME "MyVirtualHost"
public:
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    Broker(int port, const std::string &basedir)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Broker",
         muduo::net::TcpServer::kReusePort)
        ,_dispatcher(std::bind(&Broker::onUnknownMessage,
         this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
        ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
         std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
        ,_virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE))
        ,_consumer_manager(std::make_shared<ConsumerManager>())
        ,_connection_manager(std::make_shared<ConnectionManager>())
        ,_threadpool(std::make_shared<threadPool>())

    {
        //针对历史消息的所有对垒 别忘了 初始化队列的消费者结构
        QueueMap qm = _virtual_host->allQueues();
        for(auto &q : qm)
        {
            _consumer_manager->initQueueConsumer(q.first);
        }

        //注册业务请求处理函数 
        _dispatcher.registerMessageCallback<xiang11mq::openChannelRequest>(std::bind(&Broker::openChannel, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::closeChannelRequest>(std::bind(&Broker::closeChannel, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    

        _dispatcher.registerMessageCallback<xiang11mq::declareExchangeRequest>(std::bind(&Broker::declareExchange, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::deleteExchangeRequest>(std::bind(&Broker::deleteExchange, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    

        _dispatcher.registerMessageCallback<xiang11mq::declareQueueRequest>(std::bind(&Broker::declareQueue, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::deleteQueueRequest>(std::bind(&Broker::deleteQueue, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    

        _dispatcher.registerMessageCallback<xiang11mq::queueBindRequest>(std::bind(&Broker::queueBind, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::queueUnBindRequest>(std::bind(&Broker::queueUnBind, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    

        _dispatcher.registerMessageCallback<xiang11mq::basicPublishRequest>(std::bind(&Broker::basicPublish, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::basicAckRequest>(std::bind(&Broker::basicAck, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    

        _dispatcher.registerMessageCallback<xiang11mq::basicConsumerRequest>(std::bind(&Broker::basicConsumer, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<xiang11mq::basicCancelRequest>(std::bind(&Broker::basicCancel, this,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    
        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _server.setConnectionCallback(std::bind(&Broker::onConnection, this,
        std::placeholders::_1));

    }

    void start()
    {
        _server.start();    //开始监听
        _baseloop.loop();   //开始事件监控， 这是一个死循环接口
    }

private:
    //打开信道
    void openChannel(const muduo::net::TcpConnectionPtr &conn, const openChannelRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到打开信道请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("打开信道时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        return mconn->openChannel(message);

    }
    //关闭信道
    void closeChannel(const muduo::net::TcpConnectionPtr &conn, const closeChannelRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到关闭信道请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("关闭信道时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        return mconn->closeChannel(message);
    }
    //声明交换机
    void declareExchange(const muduo::net::TcpConnectionPtr &conn, const declareExchangeRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到声明交换机请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("声明交换机时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("声明交换机时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->declareExchange(message);
    }
    //删除交换机
    void deleteExchange(const muduo::net::TcpConnectionPtr &conn, const deleteExchangeRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到删除交换机请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("删除交换机时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("删除交换机时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->deleteExchange(message);
    }
    //声明队列
    void declareQueue(const muduo::net::TcpConnectionPtr &conn, const declareQueueRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到声明队列请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("声明队列时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("声明队列时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->declareQueue(message);
    }    
    //删除队列
    void deleteQueue(const muduo::net::TcpConnectionPtr &conn, const deleteQueueRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到删除队列请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("声明删除时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("声明删除时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->deleteQueue(message);
    }
    //队列绑定
    void queueBind(const muduo::net::TcpConnectionPtr &conn, const queueBindRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到队列绑定请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("队列绑定时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("队列绑定时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->queueBind(message);
    }
    //队列解绑
    void queueUnBind(const muduo::net::TcpConnectionPtr &conn, const queueUnBindRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到队列解绑请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("队列解绑时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("队列解绑时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->queueUnBind(message);
    }
    //消息发布
    void basicPublish(const muduo::net::TcpConnectionPtr &conn, const basicPublishRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到消息发布请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("发布消息时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("发布消息时, 没有找到连接对应的Channel对象");
            return;
        }
        
        return cp->basicPublish(message);
    }
    //消息确认
    void basicAck(const muduo::net::TcpConnectionPtr &conn, const basicAckRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到消息确认请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("消息确认时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("消息确认时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->basicAck(message);
    }
    //队列消息订阅
    void basicConsumer(const muduo::net::TcpConnectionPtr &conn, const basicConsumerRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到队列消息订阅请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("消息订阅时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("消息订阅时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->basicConsume(message);
    }    
    //队列消息取消订阅
    void basicCancel(const muduo::net::TcpConnectionPtr &conn, const basicCancelRequestPtr &message, muduo::Timestamp)
    {
        DLOG("收到队列消息取消订阅请求");
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if(mconn.get() == nullptr)
        {
            DLOG("取消订阅时, 没有找到连接对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if(cp.get() == nullptr)
        {
            DLOG("取消订阅时, 没有找到连接对应的Channel对象");
            return;
        }
        return cp->basicCancel(message);
    }    
    //连接建立和关闭事件
    void onConnection(const muduo::net::TcpConnectionPtr &conn)
    {
        if(conn->connected())
        {
            DLOG("新链接建立成功");
            _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);
        }
        else
        {
            _connection_manager->delConnection(conn);
            DLOG("链接关闭"); 
        }
    }
    //收到未知消息
    void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }

private:
    muduo::net::EventLoop _baseloop;
    muduo::net::TcpServer _server;                  //服务器对象
    ProtobufDispatcher _dispatcher;                 //请求分发器对象 -- 要向其中注册请求处理函数
    ProtobufCodecPtr _codec;                        //protobuf协议处理器 --针对收到的请求数据进行protobuf协议处理器
    VirtualHost::ptr _virtual_host;                 //虚拟机管理句柄
    ConsumerManager::ptr _consumer_manager;         //消费者管理句柄
    ConnectionManager::ptr _connection_manager;     //连接管理句柄
    threadPool::ptr _threadpool;                    //线程池管理句柄

};


}



#endif